package gbench.tartarus.hipponoe.stomp;

import gbench.common.tree.LittleTree.IRecord;
import gbench.tartarus.hipponoe.mvc.annotation.IRecordParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import static gbench.common.matlib.MatlibCanvas.println;
import static gbench.common.tree.LittleTree.IRecord.REC;
import static gbench.common.tree.LittleTree.MFT;
import static gbench.tartarus.hipponoe.finance.acct.AccountingApp.getAcctBalance;
import static java.lang.Thread.*;

/**
 *
 */
@RestController
public class AcctController {

    /**
     * 将客户端发送的消息广播出去
     * 从 app/publish 接收消息转发到消息队列 topic/acct_balance
     **/
    @MessageMapping("/publish")
    @SendTo("/topic/acct_balance")
    public IRecord broadcast(@IRecordParam IRecord rec) {
        println("接收到消息", rec);
        return rec;
    }

    /**
     * 启动 消息订阅系统
     * http://localhost:8082/mall/start?channel=/topic/acct_balance
     * 通过Http请求的方式触发订阅操作
     * @param channel  消息定于主体，需要以 “/topic/” 开头 比如 channel "/topic/acct_balance"
     */
    @RequestMapping("/start")
    public synchronized IRecord start(final String channel) {
        final var started = registry.getOrDefault(channel,new AtomicBoolean(false));
        if (false == started.compareAndExchange(false, true)) { // 第一次运行
            final var thd = new Thread(() -> {
                while (true) {
                    // 可以灵活设置成通道地址，实现发布订阅的功能
                    final var balance = getAcctBalance().asJavaMap();
                    simpMessagingTemplate.convertAndSend(channel, balance);
                    println(balance);
                    try {
                        sleep(10 * 1000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } // try
                }// while
            }); // Thread
            thd.setDaemon(true);
            thd.start();
            return REC("msg", MFT("{0} 成功启动",channel));
        } else {
            return REC("msg", MFT("{0} 已经启动",channel));
        }
    }

    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;
    private Map<String,AtomicBoolean> registry = new ConcurrentHashMap<String,AtomicBoolean>();
}
